package cn.antcore.kafka.builder;

import cn.antcore.kafka.ConsumerCallback;
import cn.antcore.kafka.config.KafkaConfig;
import cn.antcore.kafka.core.AntConsumer;
import cn.antcore.kafka.utils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.util.Map;

/**
 * Created By Hong on 2018/8/21
 **/
public class ConsumerBuilder {

    private static final Logger LOG = LoggerFactory.getLogger(ConsumerBuilder.class);

    private final Object target;
    private final String topic;
    private final String groupId;
    private final KafkaConfig config;

    /**
     * 创建KafkaClient
     *
     * @param target
     * @param topic            topic
     * @param groupId          groupId
     * @param bootstrapService bootstrapService
     * @param config           config
     */
    public static ConsumerBuilder create(Object target, String topic, String groupId, String bootstrapService, Class<? extends KafkaConfig> config) {
        try {
            return new ConsumerBuilder(target, topic, groupId, BeanUtils.instantiateClass(config.getConstructor(String.class), bootstrapService));
        } catch (NoSuchMethodException e) {
            LOG.error("NoSuchMethodException", e);
        }
        return new ConsumerBuilder(target, topic, groupId, BeanUtils.instantiateClass(config));
    }

    /**
     * 构造函数
     */
    private ConsumerBuilder(Object target, String topic, String groupId, KafkaConfig config) {
        this.target = target;
        this.topic = topic;
        this.groupId = groupId;
        this.config = config;
    }

    /**
     * Build
     */
    public Object build() {
        //回调类
        ConsumerCallback callback = new ConsumerCallbackImpl(target);

        Map<String, Object> map = this.config.getConsumerConfig();
        new AntConsumer<>(this.topic, this.groupId, map)
                .addCallback(callback);
        return target;
    }
}
